package com.erainm.consumeroffsetmgr;

import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.common.PartitionInfo;
import org.apache.kafka.common.TopicPartition;
import org.apache.kafka.common.serialization.IntegerDeserializer;
import org.apache.kafka.common.serialization.StringDeserializer;

import java.util.*;
import java.util.function.BiConsumer;

/**
 * Created by erainm on 2021/9/10 4:29 下午.
 *
 * @Description TODO:偏移量管理
 **/
public class MyOffsetManager {
    public static void main(String[] args) {
        Map<String, Object> configs = new HashMap<>();
        configs.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG,"lg04:9092");
        // 使用常量代替手写的字符串，配置key的反序列化器
        configs.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
        // 配置value的反序列化器
        configs.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
        // 配置消费组ID
        configs.put(ConsumerConfig.GROUP_ID_CONFIG,"mygrp1");

        KafkaConsumer<String, String> consumer = new KafkaConsumer<>(configs);

        // 订阅主题
//        consumer.subscribe(Collections.singleton("tp_demo_01"));
        // 如何手动给消费者分配分区 懒执行，可以手动分配
        // 1。 需要知道有哪些主题可以访问，消费
        // 获取当前消费者可以访问和消费的主题以及他们的分区信息
//        Map<String, List<PartitionInfo>> stringListMap = consumer.listTopics();
//        stringListMap.forEach(new BiConsumer<String, List<PartitionInfo>>() {
//            @Override
//            public void accept(String topicName, List<PartitionInfo> partitionInfos) {
//                System.out.println("主题名称：" + topicName);
//                for (PartitionInfo partitionInfo : partitionInfos) {
//                    System.out.println(partitionInfo);
//                }
//            }
//        });
        // 获取给当前消费者分配的主题分区信息
//        Set<TopicPartition> assignment1 = consumer.assignment();
//        for (TopicPartition partition1 : assignment1) {
//            System.out.println(partition1);
//        }
        System.out.println("=====================================");
        consumer.assign(Arrays.asList(
                new TopicPartition("tp_demo_01",0),
                new TopicPartition("tp_demo_01",1),
                new TopicPartition("tp_demo_01",2)
        ));

//        // 获取给当前消费者分配的主题分区信息
//        Set<TopicPartition> assignment = consumer.assignment();
//        for (TopicPartition partition : assignment) {
//            System.out.println(partition);
//        }
        // 查看当前消费者在指定主题上的消费者偏移量
//        long offset_0 = consumer.position(new TopicPartition("tp_demo_01", 0));
//        System.out.println("当前主题在0号分区上的偏移量：" + offset_0);

//        consumer.seekToBeginning(Arrays.asList(
//                new TopicPartition("tp_demo_01",0),
//                new TopicPartition("tp_demo_01",2)
//        ));
//        long offset_0 = consumer.position(new TopicPartition("tp_demo_01", 0));
//        long offset_1 = consumer.position(new TopicPartition("tp_demo_01", 1));
//        long offset_2 = consumer.position(new TopicPartition("tp_demo_01", 2));
//        System.out.println("offset_0: " + offset_0 + "\n" + "offset_1: " + offset_1 + "\n" + "offset_2: " + offset_2);
//
//        consumer.seekToEnd(Arrays.asList(new TopicPartition("tp_demo_01",0)));
//
//        offset_0 = consumer.position(new TopicPartition("tp_demo_01", 0));
//        offset_1 = consumer.position(new TopicPartition("tp_demo_01", 1));
//        offset_2 = consumer.position(new TopicPartition("tp_demo_01", 2));
//        System.out.println("offset_0: " + offset_0 + "\n" + "offset_1: " + offset_1 + "\n" + "offset_2: " + offset_2);
        consumer.seek(new TopicPartition("tp_demo_01",2),14);
        long offset_0 = consumer.position(new TopicPartition("tp_demo_01", 0));
        long offset_1 = consumer.position(new TopicPartition("tp_demo_01", 1));
        long offset_2 = consumer.position(new TopicPartition("tp_demo_01", 2));
        System.out.println("offset_0: " + offset_0 + "\n" + "offset_1: " + offset_1 + "\n" + "offset_2: " + offset_2);


        consumer.close();
    }
}
